package org.pupil.rpc.protocol.codec;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.pupil.rpc.common.constants.MsgType;
import org.pupil.rpc.common.constants.ProtocolConstants;
import org.pupil.rpc.protocol.MsgHeader;
import org.pupil.rpc.protocol.RpcProtocol;
import org.pupil.rpc.protocol.RpcRequest;
import org.pupil.rpc.protocol.RpcResponse;
import org.pupil.rpc.protocol.compress.Compress;
import org.pupil.rpc.protocol.compress.CompressFactory;
import org.pupil.rpc.protocol.serialization.RpcSerialization;
import org.pupil.rpc.protocol.serialization.SerializationFactory;

import java.util.List;

/**
 * 解码器
 */
public class RpcDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 如果可读字节数少于协议头长度，说明还没有接收完整个协议头，直接返回
        if (in.readableBytes() < ProtocolConstants.HEADER_TOTAL_LEN) {
            return;
        }

        // 标记当前读取位置，便于后面回退
        in.markReaderIndex();

        short magic = in.readShort(); // 魔数
        if (magic != ProtocolConstants.MAGIC) {
            throw new IllegalArgumentException("magic number is illegal, " + magic);
        }

        byte version = in.readByte(); // 协议版本号
        byte msgType = in.readByte(); // 数据类型
        byte status = in.readByte(); // 状态
        long requestId = in.readLong(); // 请求 ID

        final int serLen = in.readInt(); // 序列化规则字节数组长度
        // 可读数据小于序列化规则字节数组长度，则数据没有获取完，重新读
        if (in.readableBytes() < serLen) {
            in.resetReaderIndex();
            return;
        }
        // 获取序列化规则字节数组
        final byte[] ser = new byte[serLen];
        in.readBytes(ser);
        final String serialization = new String(ser);

        final int compressLen = in.readInt(); // 压缩规则字节数组长度
        // 可读数据小于压缩规则字节数组长度，则数据没有获取完，重新读
        if(in.readableBytes() < compressLen){
            in.resetReaderIndex();
            return;
        }
        // 获取压缩规则字节数组
        final byte[] compress = new byte[compressLen];
        in.readBytes(compress);
        final String compression = new String(compress);

        int msgLen = in.readInt(); // 数据长度
        // 可读数据小于数据长度，则数据没有获取完，重新读
        if (in.readableBytes() < msgLen) {
            in.resetReaderIndex();
            return;
        }
        // 获取消息体  字节数组
        byte[] data = new byte[msgLen];
        in.readBytes(data);

        // 处理消息类型
        MsgType msgTypeEnum = MsgType.findByType(msgType);
        if (msgTypeEnum == null) {
            // 非法消息类型，直接返回
            return;
        }

        // 构建消息头
        MsgHeader header = new MsgHeader();
        header.setMagic(magic);
        header.setVersion(version);
        header.setStatus(status);
        header.setRequestId(requestId);
        header.setMsgType(msgType);
        header.setSerializations(ser);
        header.setSerializationLen(serLen);
        header.setCompression(compress);
        header.setCompressLen(compressLen);
        header.setMsgLen(msgLen);
        // 获取序列化器
        RpcSerialization rpcSerialization = SerializationFactory.get(serialization);
        //获取压缩器
        Compress rpcCompression = CompressFactory.get(compression);
        data = rpcCompression.decompress(data);
        // 根据消息类型进行处理(如果消息类型过多可以使用策略+工厂模式进行管理)
        switch (msgTypeEnum) {
            // 请求消息
            case REQUEST:
                RpcRequest request = rpcSerialization.deSerialize(data, RpcRequest.class);
                if (request != null) {
                    RpcProtocol<RpcRequest> protocol = new RpcProtocol<>();
                    protocol.setHeader(header);
                    protocol.setBody(request);
                    out.add(protocol);
                }
                break;
            // 响应消息
            case RESPONSE:
                RpcResponse response = rpcSerialization.deSerialize(data, RpcResponse.class);
                if (response != null) {
                    RpcProtocol<RpcResponse> protocol = new RpcProtocol<>();
                    protocol.setHeader(header);
                    protocol.setBody(response);
                    out.add(protocol);
                }
                break;
        }
    }
}
